package com.sanyue.codingsettlement.mqstudy.rabbitmq.publisherConfirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 批量确认（发布者确认）
 * @author: Guan
 **/
public class AsynConfirm {
    public static void main(String[] args) throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //开启发布者确认
        channel.confirmSelect();
        //创建回调函数用于异步确认
        ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {
            System.out.println("发布的消息已经成功到达队列");
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(
                        sequenceNumber, true
                );
                confirmed.clear();
            } else {
                outstandingConfirms.remove(sequenceNumber);
            }
        };

        channel.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> {
            String body = outstandingConfirms.get(sequenceNumber);
            System.err.format(
                    "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
                    body, sequenceNumber, multiple
            );
            cleanOutstandingConfirms.handle(sequenceNumber, multiple);
        });

        //发布消息
        for(int i = 0;i < 5;i++){
            String message = "消息" + i;
            String queueName = channel.queueDeclare().getQueue();
            //1.将所有的序列号和消息存放到map中
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", "queueName", null, message.getBytes());
        }
    }
}
